package org.niit.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object CreateRDD01 {




  def main(args: Array[String]): Unit = {
      //Spark:基于内存式的计算框架  RDD:弹性分布式数据集
    //1.准备环境
       // -- 创建Spark的配置                             CPU核数等于分区数
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark") //重点
      //自定义设置分区数
    sparkConf.set("spark.default.parallelism","4")
    //2.创建SparkContext 去加载配置
    val sparkContext = new SparkContext(sparkConf)//重点
    //设置日志级别
    sparkContext.setLogLevel("ERROR")

    //3.1从内存中创建RDD
    //memoryCreateRDD(sparkContext)

    //3.2从外部（文件）创建RDD
    fileCreateRDD(sparkContext)
    //4.关闭环境
    sparkContext.stop()
  }

  def memoryCreateRDD(sparkContext: SparkContext) = {
    //创建RDD--内存中创建RDD,将内存中集合的数据作为分析(处理)数据源
      //RDD 就可以理解 List 只不过方法写法上不一样，比Listduo多了很多方法
    val rdd1: RDD[Int] = sparkContext.parallelize(1 to 10) //1-10 RDD  == RDD就是一个自带很多方法的 列表
    val rdd2: RDD[Int] = sparkContext.parallelize(1 to 10,3)//3:分区数
    println(rdd1.getNumPartitions)
    println(rdd2.getNumPartitions)
    //将计算结果保存在文件当中
   //rdd1.saveAsTextFile("output")
    //makeRDD 和  parallelize 是一样，parallelize是makeRDD的底层写法
   val rdd3: RDD[Int] = sparkContext.makeRDD(1 to 10,6)  // 重点
    //rdd进行迭代
    val ints: Array[Int] = rdd3.collect() //先将rdd变成数组   List.toArray  1步
    ints.foreach( println ) // 2步
    //   数组    |   对数组的遍历
    rdd3.collect().foreach( println ) //合并

  }

  def fileCreateRDD(sparkContext: SparkContext) = {
    //1.在项目目录中创建一个input文件夹，并在该文件夹中创建1.txt目录，里面写一些东西
    val rdd4: RDD[String] = sparkContext.textFile("input/1.txt")
    //遍历rdd
    rdd4.collect().foreach( println )
  }

}
